package org.budo.canal.logposition.zookeeper;

import java.util.ArrayList;
import java.util.Collections;
import java.util.HashSet;
import java.util.List;
import java.util.Map;
import java.util.Set;
import java.util.concurrent.Executors;
import java.util.concurrent.ScheduledExecutorService;
import java.util.concurrent.TimeUnit;

import org.I0Itec.zkclient.exception.ZkNoNodeException;
import org.budo.canal.logposition.BudoLogPositionManager;
import org.budo.support.lang.util.MapUtil;
import org.budo.support.lang.util.ReflectUtil;

import com.alibaba.otter.canal.common.utils.JsonUtils;
import com.alibaba.otter.canal.common.zookeeper.ZkClientx;
import com.alibaba.otter.canal.common.zookeeper.ZookeeperPathUtils;
import com.alibaba.otter.canal.parse.exception.CanalParseException;
import com.alibaba.otter.canal.parse.index.AbstractLogPositionManager;
import com.alibaba.otter.canal.parse.index.MemoryLogPositionManager;
import com.alibaba.otter.canal.protocol.position.LogPosition;

import lombok.Getter;
import lombok.Setter;
import lombok.extern.slf4j.Slf4j;

/**
 * @author lmw
 * 
 * @see com.alibaba.otter.canal.parse.index.PeriodMixedLogPositionManager
 */
@Slf4j
@Getter
@Setter
public class BudoPeriodMixedZooKeeperLogPositionManager extends AbstractLogPositionManager implements BudoLogPositionManager {
    private String pathPrefix = "";

    private ZkClientx zkClientx;

    private MemoryLogPositionManager memoryLogPositionManager;

    private Long period;

    private Set<String> persistDestinationTasks;

    private ScheduledExecutorService executorService;

    public BudoPeriodMixedZooKeeperLogPositionManager() {
        this.memoryLogPositionManager = new MemoryLogPositionManager();
        this.period = 1000L;
        this.persistDestinationTasks = Collections.synchronizedSet(new HashSet<String>());
        this.executorService = Executors.newScheduledThreadPool(1);
    }

    public BudoPeriodMixedZooKeeperLogPositionManager(ZkClientx zkClient) {
        this();

        this.zkClientx = zkClient;
    }

    /**
     * 读时内存中有就用内存的,没有就读ZK
     */
    @Override
    public LogPosition getLatestIndexBy(String destination) {
        LogPosition fromMem = memoryLogPositionManager.getLatestIndexBy(destination);

        if (null != fromMem && null != fromMem.getPostion() && null != fromMem.getIdentity()) {
            return fromMem;
        }

        LogPosition fromZk = this.getFromZk(destination);
        log.warn("#72 getLatestIndexBy return fromZk, destination=" + destination + ", fromMem=" + fromMem + ", fromZk=" + fromZk);

        return fromZk;
    }

    /**
     * 写时先写内存,然后异步写到ZK
     */
    @Override
    public void persistLogPosition(String destination, LogPosition logPosition) throws CanalParseException {
        persistDestinationTasks.add(destination);

        memoryLogPositionManager.persistLogPosition(destination, logPosition);
    }

    @Override
    public void start() {
        super.start();

        memoryLogPositionManager.start();

        // 启动定时工作任务
        executorService.scheduleAtFixedRate(new Runnable() {
            public void run() {
                Set<String> persistDestinationTasks = BudoPeriodMixedZooKeeperLogPositionManager.this.getPersistDestinationTasks();
                List<String> destinations = new ArrayList<String>(persistDestinationTasks);

                for (String destination : destinations) {
                    try { // 定时将内存中的最新值刷到zookeeper中，多次变更只刷一次
                        LogPosition logPosition = BudoPeriodMixedZooKeeperLogPositionManager.this.getLatestIndexBy(destination);
                        BudoPeriodMixedZooKeeperLogPositionManager.this.persistToZk(destination, logPosition); // 写到ZK
                        persistDestinationTasks.remove(destination);
                    } catch (Throwable e) {
                        log.error("#88 period update" + destination + " curosr failed!", e);
                    }
                }
            }
        }, period, period, TimeUnit.MILLISECONDS);
    }

    @Override
    public void stop() {
        super.stop();

        memoryLogPositionManager.stop();

        executorService.shutdown();
    }

    private void persistToZk(String destination, LogPosition logPosition) {
        String path = this.getPathPrefix() + ZookeeperPathUtils.getParsePath(destination);
        byte[] data = JsonUtils.marshalToByte(logPosition);

        try {
            this.getZkClientx().writeData(path, data);
        } catch (ZkNoNodeException e) {
            this.getZkClientx().createPersistent(path, data, true);
        }
    }

    private LogPosition getFromZk(String destination) {
        String path = this.getPathPrefix() + ZookeeperPathUtils.getParsePath(destination);
        byte[] data = this.getZkClientx().readData(path, true);
        if (data == null || data.length == 0) {
            return null;
        }

        return JsonUtils.unmarshalFromByte(data, LogPosition.class);
    }

    @Override
    public Boolean deleteLogPosition(String destination) {
        Map<String, LogPosition> positions = (Map<String, LogPosition>) ReflectUtil.getFieldValue(MemoryLogPositionManager.class, "positions", this.memoryLogPositionManager);
        MapUtil.remove(positions, destination);

        String path = this.getPathPrefix() + ZookeeperPathUtils.getParsePath(destination);
        return this.getZkClientx().delete(path);
    }
}
